module dactor.actor;

private import dactor.coroutine;
private import dactor.eventlet;
private import dactor.scheduler;
private import dactor.dist;
private import dactor.log;

private import core.thread;

/**
 * Sleep but no block
 */
void sleep(int timeout=-1) {
    auto e = new TimerEvent(timeout);
    e.wait();
}

/**
 * Send message to other process by pid.
 */
void send(Pid pid, Object o) {
    if (pid.node) {
        return sendRemoteMessage(pid, o);
    }

    auto coroutine = Scheduler.instance.findByPid(pid);
    if (!coroutine) {
        logDebug("target %s not exist", pid);
        return;
    }

    sendMessage(coroutine, o);
}

/**
 * Send message to other process by name.
 */
void send(string receiver, Object o) {
    int pos = std.string.find(receiver, '@');
    if (pos >= 0) {
        return sendRemoteMessage(receiver[0 .. pos], receiver[pos+1 .. $], o);
    }

    auto coroutine = Scheduler.instance.findByName(receiver);
    if (!coroutine) {
        logDebug("target %s not exist", receiver);
        return;
    }

    sendMessage(coroutine, o);
}

private
void sendMessage(Coroutine coroutine, Object o) {
    Message msg = new Message;
    msg.from = self().pid;
    msg.to = coroutine.pid;
    msg.message = o;
    doSendMessage(coroutine, msg);
}

private
void doSendMessage(Coroutine coroutine, Message msg) {
    coroutine.addMessage(msg);
    if (coroutine.status == Coroutine.Status.RECEIVING) {
        Scheduler.instance.schedule(coroutine);
    }
}

private
void sendRemoteMessage(Pid pid, Object o) {
    Pid proxyPid = NodeManager.instance.findProxy(pid.node);
    if (proxyPid == Pid.INVALID_PID) {
        return;
    }

    Coroutine coroutine = Scheduler.instance.findByPid(proxyPid);
    if (!coroutine) {
        return;
    }

    Message msg = new Message;
    msg.from = self().pid;
    msg.to = pid;
    msg.message = o;
    doSendMessage(coroutine, msg);
}

private
void sendRemoteMessage(string name, string node, Object o) {
    logDebug("send remote message to: %s@%s, message: %s", name, node, o);

    Pid proxyPid = NodeManager.instance.findProxy(node);
    if (proxyPid == Pid.INVALID_PID) {
        logDebug("receiver %s not found", node);
        return;
    }

    Coroutine coroutine = Scheduler.instance.findByPid(proxyPid);
    if (!coroutine) {
        logDebug("proxy of %s not found", node);
        return;
    }

    Message msg = new Message;
    msg.from = self().pid;
    msg.toName = name;
    msg.message = o;
    doSendMessage(coroutine, msg);
}

/**
 * Receive message with timeout.
 */
Message receive(int timeout=-1) {

    Coroutine coroutine = Scheduler.instance.getCurrent();
    if (Message m = coroutine.getMessage()) {
        return m;
    }

    // no messages
    if (timeout == 0)
        return null;

    // wait message
    coroutine.status = Coroutine.Status.RECEIVING;
    if (timeout == -1) {
        Fiber.yield();
    } else {
        auto e = new TimerEvent(timeout);
        e.wait();
    }

    // wake up
    return coroutine.getMessage();
}

